# -*- coding: utf-8 -*-
"""
Kafka消息发送API视图

该模块提供Kafka消息发送相关的API接口，包括：
1. HelloWorld测试接口
2. 发送消息到Kafka主题

主要功能：
- 创建Kafka生产者连接
- 发送JSON格式消息到指定主题
- 处理发送结果和异常情况
- 提供统一的API响应格式
"""

from django.shortcuts import render
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
import json
from kafka import KafkaProducer
from kafka.errors import KafkaError
import logging

# 配置日志记录器
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Create your views here.
def hello_world(request):
    """
    HelloWorld测试接口
    
    用于测试API服务是否正常运行。
    
    Returns:
        JsonResponse: 包含问候消息和状态的JSON响应
    """
    return JsonResponse({
        'message': 'Hello, World!',
        'status': 'success'
    })

@csrf_exempt
def send_kafka_message(request):
    """
    发送Kafka消息的API接口
    
    该接口接收JSON格式的请求数据，创建Kafka生产者连接，
    并将消息发送到指定的Kafka主题。
    
    Args:
        request: HTTP请求对象，包含以下JSON参数：
            - kafkaUrl (str): Kafka服务器地址（如：localhost:9092）
            - topic (str): 目标主题名称
            - msg (any): 要发送的消息内容（将被序列化为JSON）
    
    Returns:
        JsonResponse: 包含以下字段的JSON响应：
            - code (int): 状态码，0表示成功，1表示失败
            - message (str): 响应消息描述
    
    Raises:
        json.JSONDecodeError: 当请求体不是有效JSON格式时
        KafkaError: 当Kafka操作失败时
        Exception: 其他未预期的异常
    """
    # 检查请求方法，只允许POST请求
    if request.method != 'POST':
        return JsonResponse({
            'code': 1,
            'message': '仅支持POST请求'
        })
    
    try:
        # 解析请求体中的JSON数据
        data = json.loads(request.body)
        kafka_url = data.get('kafkaUrl')  # Kafka服务器地址
        topic = data.get('topic')         # 目标主题名称
        message = data.get('msg')         # 要发送的消息内容
        
        # 校验必填参数是否都已提供
        if not all([kafka_url, topic, message]):
            return JsonResponse({
                'code': 1,
                'message': '缺少必要参数：kafkaUrl, topic或msg'
            })
        
        # 创建Kafka生产者实例
        # bootstrap_servers: 指定Kafka集群的服务器地址列表
        # value_serializer: 消息值的序列化器，将Python对象转换为JSON字节串
        producer = KafkaProducer(
            bootstrap_servers=[kafka_url],
            value_serializer=lambda v: json.dumps(v).encode('utf-8')
        )
        
        # 异步发送消息到指定主题
        # 返回一个Future对象，可用于获取发送结果
        future = producer.send(topic, message)
        
        # 同步等待发送完成，最多等待10秒
        # 返回RecordMetadata对象，包含分区和偏移量信息
        record_metadata = future.get(timeout=10)
        
        # 关闭生产者连接，释放资源
        producer.close()
        
        # 记录发送成功的日志，包含主题、分区和偏移量信息
        logger.info(f"消息发送成功: topic={topic}, partition={record_metadata.partition}, offset={record_metadata.offset}")
        
        # 返回成功响应
        return JsonResponse({
            'code': 0,
            'message': '消息发送成功'
        })
    
    except json.JSONDecodeError:
        # 处理JSON解析错误，通常是请求体格式不正确
        return JsonResponse({
            'code': 1,
            'message': '无效的JSON格式'
        })
    except KafkaError as e:
        # 处理Kafka相关错误，如连接失败、主题不存在等
        logger.error(f"Kafka错误: {str(e)}")
        return JsonResponse({
            'code': 1,
            'message': f'Kafka错误: {str(e)}'
        })
    except Exception as e:
        # 处理其他未预期的异常
        logger.error(f"发生错误: {str(e)}")
        return JsonResponse({
            'code': 1,
            'message': f'发生错误: {str(e)}'
        })
